package com.ubox.mqtt_upush.mqttbroker;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.activemq.broker.Broker;
import org.apache.activemq.broker.BrokerFilter;
import org.apache.activemq.broker.ConnectionContext;
import org.apache.activemq.broker.ProducerBrokerExchange;
import org.apache.activemq.broker.region.Subscription;
import org.apache.activemq.command.ConnectionInfo;
import org.apache.activemq.command.ConsumerId;
import org.apache.activemq.command.ConsumerInfo;
import org.apache.activemq.command.Message;
import org.apache.activemq.command.ProducerInfo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.ubox.mqtt_upush.MqttUpushApplication;
import com.ubox.mqtt_upush.utils.JsonUtil;

public class SessionFilter extends BrokerFilter {

	private final static Logger logger = LoggerFactory.getLogger(SessionFilter.class);
	// clientId->ConnectionContext
	protected final Map<String, ConnectionContext> clientIdConnection = new ConcurrentHashMap<>();
	// ConsumerId->ConnectionContext
	protected Map<ConsumerId, ConnectionContext> connetionIdConnection = new ConcurrentHashMap<>();

	private MessageDispatcher messageTemplate;

	public boolean isOnline(String vmCode) {
		ConnectionContext connectionState = clientIdConnection.get(vmCode);
		if (connectionState == null) {
			return false;
		} else {
			return true;
		}
	}

	public SessionFilter(Broker next) {
		super(next);
	}

	@Override
	public void addConnection(ConnectionContext context, ConnectionInfo info) throws Exception {

		if (context.getConnector().toString().equals("mqtt+nio")) {
			getNext().addConnection(context, info);
			logger.info("mqtt connected {} {} {}", info.getUserName(), info.getClientId(), info.getClientIp());
			return;
		}
        //SSL登录
		getNext().addConnection(context, info);

		clientIdConnection.put(info.getClientId(), context);

		logger.info("mqtt addConnection {} {}", info.getClientId(), info.getClientIp());
	}

	@Override
	public void addProducer(ConnectionContext context, ProducerInfo info) throws Exception {
		getNext().addProducer(context, info);
	}

	@Override
	public void removeConnection(ConnectionContext context, ConnectionInfo info, Throwable error) throws Exception {
		clientIdConnection.remove(info.getClientId());
		logger.info("mqtt removeConnection {} {}", info.getClientId(), info.getClientIp());
		getNext().removeConnection(context, info, error);
	}

	@Override
	public void send(ProducerBrokerExchange producerExchange, Message messageSend) throws Exception {
		String msg = new String(messageSend.getMessage().getContent().data);

		logger.info("[{}] mqtt send qos={} {} msg: {} ", producerExchange.getConnectionContext().getClientId(),
				JsonUtil.toString(messageSend.getProperties().get("ActiveMQ.MQTT.QoS")), messageSend.getDestination(),
				msg);
		if (messageTemplate == null) {
			messageTemplate = MqttUpushApplication.getApplicationContext().getBean(MessageDispatcher.class);
		}

		getNext().send(producerExchange, messageSend);

		if (messageSend.getDestination().getPhysicalName().equals("v1.cloud.request")) {
			String innerCode = producerExchange.getConnectionContext().getClientId();
			ConnectionContext session = producerExchange.getConnectionContext();
			messageTemplate.dispatcher(session, innerCode, msg);
		}
	}

	@Override
	public Subscription addConsumer(ConnectionContext context, ConsumerInfo info) throws Exception {
		connetionIdConnection.put(info.getConsumerId(), context);
		return getNext().addConsumer(context, info);
	}

	@Override
	public void removeConsumer(ConnectionContext context, ConsumerInfo info) throws Exception {
		connetionIdConnection.remove(info.getConsumerId());
		getNext().removeConsumer(context, info);
	}
}
